package com.myCompany.edu.realtime.app.dws;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.google.inject.internal.cglib.core.$LocalVariablesSorter;
import com.google.inject.internal.cglib.proxy.$Factory;
import com.myCompany.edu.realtime.app.BaseAppV1;
import com.myCompany.edu.realtime.bean.TrafficPageViewBean;
import com.myCompany.edu.realtime.common.Constant;
import com.myCompany.edu.realtime.util.AtguiguUtil;
import com.myCompany.edu.realtime.util.FlinkSinkUtil;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.FilterFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.api.common.functions.RichReduceFunction;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.streaming.api.functions.windowing.ProcessAllWindowFunction;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.assigners.TumblingProcessingTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;
import org.apache.kafka.common.protocol.types.Field;

import java.time.Duration;
import java.util.Iterator;

/**
 * @author
 * @date 2022/10/21 11:06
 */

public class Dws01_DwsTrafficSourceAreaProvinceUViewWindow extends BaseAppV1 {
    public static void main(String[] args) {
        new Dws01_DwsTrafficSourceAreaProvinceUViewWindow().init(3001, 2,
                "Dws01_DwsTrafficSourceAreaProvinceUViewWindow", Constant.TOPIC_DWD_TRAFFIC_DISPLAY);
    }

    @Override
    protected void handle(StreamExecutionEnvironment env, DataStreamSource<String> stream) {
        SingleOutputStreamOperator<TrafficPageViewBean> beanStream = parseBean(stream);
//        beanStream.print();
        SingleOutputStreamOperator<TrafficPageViewBean> resultBean = windowAndAgg(beanStream);
        resultBean.print();
        writeClickHouse(resultBean);
    }

    private void writeClickHouse(SingleOutputStreamOperator<TrafficPageViewBean> resultBean) {
        resultBean.addSink(FlinkSinkUtil.getClickHouseSink("dws_traffic_source_area_province_user_view_window", TrafficPageViewBean.class));
    }

    private SingleOutputStreamOperator<TrafficPageViewBean> windowAndAgg(SingleOutputStreamOperator<TrafficPageViewBean> stream) {
        return stream
                .assignTimestampsAndWatermarks(
                        WatermarkStrategy
                                .<TrafficPageViewBean>forBoundedOutOfOrderness(Duration.ofSeconds(3))
                                .withTimestampAssigner((bean, ts) -> bean.getTs())
                )
                .keyBy(new KeySelector<TrafficPageViewBean, String>() {
                    @Override
                    public String getKey(TrafficPageViewBean value) throws Exception {

                        return value.getCh() + value.getAr() + value.getIsNew();
                    }
                })
                .window(TumblingEventTimeWindows.of(Time.seconds(5)))
                .reduce(new ReduceFunction<TrafficPageViewBean>() {
                            @Override
                            public TrafficPageViewBean reduce(TrafficPageViewBean bean1, TrafficPageViewBean bean2) throws Exception {
                                bean1.setUvCt(bean1.getUvCt() + bean2.getUvCt());
                                bean1.setSvCt(bean1.getSvCt() + bean2.getSvCt());
                                bean1.setPvCt(bean1.getPvCt() + bean2.getPvCt());
                                bean1.setSkipCt(bean1.getSkipCt() + bean2.getSkipCt());
                                bean1.setDurSum(bean1.getDurSum() + bean2.getDurSum());

                                return bean1;
                            }
                        },
                        new ProcessWindowFunction<TrafficPageViewBean, TrafficPageViewBean, String, TimeWindow>() {
                            @Override
                            public void process(String s, Context context, Iterable<TrafficPageViewBean> iterable, Collector<TrafficPageViewBean> collector) throws Exception {
                                TrafficPageViewBean bean = iterable.iterator().next();
                                bean.setStt(AtguiguUtil.toDateTime(context.window().getStart()));
                                bean.setEdt(AtguiguUtil.toDateTime(context.window().getEnd()));
                                collector.collect(bean);
                            }
                        }
                );
    }



   private SingleOutputStreamOperator<TrafficPageViewBean> parseBean(DataStreamSource<String> stream) {
        return stream
                .keyBy(value -> JSON.parseObject(value).getString("uid"))
                .map(new RichMapFunction<String, TrafficPageViewBean>() {
                    ValueState<String> uidState;
                    ValueState<String> dayState;
                    ValueState<String> sessionIdState;
                    ValueState<Long> durSumState;
                    ValueState<Long> flagState;
                    ValueState<String> pageIdState;

                    @Override
                    public void open(Configuration parameters) throws Exception {
                        uidState = getRuntimeContext().getState(new ValueStateDescriptor<String>(
                                "uidState", String.class));
                        dayState = getRuntimeContext().getState(new ValueStateDescriptor<String>(
                                "dayState", String.class));
                        sessionIdState = getRuntimeContext().getState(new ValueStateDescriptor<String>(
                                "sessionIdState", String.class));
                        durSumState = getRuntimeContext().getState(new ValueStateDescriptor<Long>(
                                "durSumState", Long.class));
                        flagState = getRuntimeContext().getState(new ValueStateDescriptor<Long>(
                                "flagState", Long.class));
                        pageIdState = getRuntimeContext().getState(new ValueStateDescriptor<String>(
                                "pageIdState", String.class));
                    }

                    @Override
                    public TrafficPageViewBean map(String value) throws Exception {

                        String yesterday = dayState.value();
                        String sessionId = sessionIdState.value();
                        Long flag = flagState.value();
                        String pageIdSta = pageIdState.value();


                        JSONObject common = JSON.parseObject(value);
                        String ch = common.getString("ch");
                        String ar = common.getString("ar");

                        Long uvCt = 0L;
                        Long svCt = 0L;
                        Long pvCt = 0L;
                        Long skipCt = 0L;

                        String today = AtguiguUtil.toDate(common.getLong("ts"));
                        String sid = common.getString("sid");
                        String pageId = common.getString("page_id");
                        Long duringTime = common.getLong("during_time");
                        Long isNew = common.getLong("is_new");
                        String uid = common.getString("uid");


                        if (yesterday != today) {
                            uvCt = 1L;
                            dayState.update(today);
                        }

//                        System.out.println("sid: " + sid + "\t\t\t\t\t" + "sessionId: " + sessionId + "\t\t\t\t" + "uid: " +uid);

                        if (!sid.equals(sessionId)) {
                            svCt = 1L;
                            if (!pageId.equals(pageIdSta)) {
                                skipCt = 1L;
                                pvCt = 1L;
                                pageIdState.update(pageId);
                                flagState.update(1L);
                            } else {
                                if (flag == 1) {
                                    skipCt = -1L;
                                    flagState.update(-1L);
                                } else {
                                    skipCt = 0L;
                                    flagState.update(0L);
                                }
                            }
                            sessionIdState.update(sid);
                        }


//                        System.out.println("pageId: " + sid + "\t\t\t\t\t" + "pageIdSta: " + sessionId + "\t\t\t\t" + "uid: " +uid);
                        return new TrafficPageViewBean("", "", ch, ar, isNew,
                                uvCt, svCt, pvCt, skipCt, duringTime, common.getLong("ts"));
                    }
                });

    }
}
